//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// hash_join_executor.cpp
//
// Identification: src/execution/hash_join_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

/*
 * 实现了一个哈希链接操作。该操作将来自其两个子执行器的记录组合在一起。
 * */
#include "execution/executors/hash_join_executor.h"

namespace bustub {

HashJoinExecutor::HashJoinExecutor(ExecutorContext *exec_ctx, const HashJoinPlanNode *plan,
                                   std::unique_ptr<AbstractExecutor> &&left_child,
                                   std::unique_ptr<AbstractExecutor> &&right_child)
    : AbstractExecutor(exec_ctx)
    ,plan_(plan)
    ,left_child_executor_(std::move(left_child))
    ,right_child_executor_(std::move(right_child)){}

/*
 *  和之前的loop_join一样，先在init中保存结果，然后在next中对结果进行保存。
 *  对于两张表而言，符合同一哈希结果的行可能有多个。
 *      所以选择在哈系表中保存由tuple组成的vector,来存储左表中同一哈希结果的多行数据
 *
 *   思路：先将左表中的全部列根据join条件散列到一张哈希表中，然后遍历右表，
 *          右表中的每行数据都要根据join条件的哈希结果在之前得到的哈希表中找到所有行，并一一组合，
 *                  这样得到结果才是笛卡尔积
 *
 * */
//void HashJoinExecutor::Init() {
//  left_child_executor_->Init();
//  right_child_executor_->Init();
//
//  // 通过左侧查询构造哈希
//  Tuple left_tuple;
//  RID left_rid;
//  while(left_child_executor_->Next(&left_tuple,&left_rid)){
//    // 构造key
//    HashJoinKey dis_key;
//    dis_key.column_value_ = plan_->LeftJoinKeyExpression()->Evaluate(&left_tuple,
//                      left_child_executor_->GetOutputSchema());
//    if(map_.count(dis_key)!=0){
//      map_[dis_key].emplace_back(left_tuple);
//    }else{
//      map_[dis_key]= std::vector{left_tuple};
//    }
//  }
//  // 遍历右侧查询，得到查询结果
//  Tuple right_tuple;
//  RID right_rid;
//  while(right_child_executor_->Next(&right_tuple,&right_rid)){
//    // 构造key
//    HashJoinKey dis_key;
////    dis_key.column_value_ = plan_->RightJoinKeyExpression()->
//    if(map_.count(dis_key) !=0){
//      for(auto &left_tuple_new:map_.find(dis_key)->second){
//        std::vector<Value> output;
//        for(const auto &col:GetOutputSchema()->GetColumns()){
//          output.push_back(col.GetExpr()->EvaluateJoin(&left_tuple_new,left_child_executor_->GetOutputSchema(),
//                                                       &right_tuple,right_child_executor_->GetOutputSchema()));
//        }
//        result_.emplace_back(Tuple(output,GetOutputSchema()));
//      }
//    }
//  }
//}

void HashJoinExecutor::Init() {
  left_child_executor_->Init();
  right_child_executor_->Init();
  // 通过左侧查询构造哈希
  Tuple left_tuple;
  RID left_rid;
  while (left_child_executor_->Next(&left_tuple, &left_rid)) {
    // 构造Key
    HashJoinKey dis_key;
    dis_key.column_value_ =
        plan_->LeftJoinKeyExpression()->Evaluate(&left_tuple, left_child_executor_->GetOutputSchema());
    // 重复的数据要额外保存
    if (map_.count(dis_key) != 0) {
      map_[dis_key].emplace_back(left_tuple);
    } else {
      map_[dis_key] = std::vector{left_tuple};
    }
  }
  // 遍历右侧查询，得到查询结果
  Tuple right_tuple;
  RID right_rid;
  while (right_child_executor_->Next(&right_tuple, &right_rid)) {
    // 构造Key
    HashJoinKey dis_key;
    dis_key.column_value_ =
        plan_->RightJoinKeyExpression()->Evaluate(&right_tuple, right_child_executor_->GetOutputSchema());
    // 遍历每一个对应的左侧查询
    if (map_.count(dis_key) != 0) {
      for (auto &left_tuple_new : map_.find(dis_key)->second) {
        std::vector<Value> output;
        for (const auto &col : GetOutputSchema()->GetColumns()) {
          output.push_back(col.GetExpr()->EvaluateJoin(&left_tuple_new, left_child_executor_->GetOutputSchema(),
                                                       &right_tuple, right_child_executor_->GetOutputSchema()));
        }
        result_.emplace_back(Tuple(output, GetOutputSchema()));
      }
    }
  }
}
bool HashJoinExecutor::Next(Tuple *tuple, RID *rid) {
  if(now_id_ < result_.size()){
    *tuple =  result_[now_id_];
    *rid = tuple->GetRid();
    now_id_++;
    return true;
  }
  return false;
}

}  // namespace bustub
